package com.fsl.springbootstreamkafka.original;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.*;


/**
 * @author dell
 * kafka 原生API调用 consumer
 */
public class KafkaConsumerTest {

    public static void main(String[] args) {

        Properties props = new Properties();
        /* 定义kakfa 服务的地址，不需要将所有broker指定上 */
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        /* 制定consumer group */
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test1");
        /* 是否自动确认offset */
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /* 自动确认offset的时间间隔 */
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        /* key的序列化类 */
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /* value的序列化类 */
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        /*
        当kafka中没有指定offset初值时，或指定的offset不存在时，从这里读取offset的值。其取值的意义为：
        earliest:指定offset为第一条offset
        latest: 指定offset为最后一条offset
        */
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        /* 定义consumer */
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        /* 消费者订阅的topic, 可同时订阅多个 */
        consumer.subscribe(Collections.singletonList("mytopic"));

        List<ConsumerRecord<String, String>> buffer = new ArrayList<>();

        Integer minBatchSize = 5;

        /* 读取数据，读取超时时间为100ms */
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(5*1000));
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("offset------->" + record.offset() + " ,key------>" + record.key() + " ,value------->" + record.value());
            buffer.add(record);
            /* 数据达到批量要求，就写入DB，同步确认offset */
            if (buffer.size() >= minBatchSize) {
                //consumer.commitSync();
                //buffer.clear();
                System.out.println("current size ------>");
                buffer.clear();
            }
        }
        /*
        * 读取之后发现offset每次都是从60开始，这个是为什么呢？
        * 解答：
        *     频繁的执行这个main方法的时候，第一次会能够消费掉消息，然后第二次就不会从0开始消费消息了
        *     然后间隔一段时间，再来执行这个main方法，有能够从0开始消费消息了
        * 猜想：
        *     应该是kafka对消费者有相关的设置，能够识别是不是同一个请求(短时内重复请求就是同一个请求)
        *     假如是同一个请求，那么之前消费过的消息就不会再重复消费了
        */
    }

}
